package pub.dsb.example.rx.operators;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;

import java.util.Arrays;
import java.util.concurrent.*;

/**
 * <p>
 *
 * </p>
 *
 * @author Yi
 * @version V1.0.0
 * @date 2020/12/24 10:02
 * @modificationHistory=========================逻辑或功能性重大变更记录
 * @modify By: {修改人} 2020/12/24 10:02
 * @modify reason: {方法名}:{原因}
 * ...
 * @since V1.0.0
 */
public class ObservableOperatorTest {

    /**
     * 操作符
     *
     * @param args
     */
    public static void main(String[] args) {
        System.out.println("DEMO 1 ---------------");
        Observable.just(1, 2, 3)
                .subscribe(new Observer<Integer>() {

                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        d.dispose();
                        System.out.println("Cancel !");
                    }

                    @Override
                    public void onNext(Integer item) {
                        System.out.println("Next: " + item);
                    }

                    @Override
                    public void onError(Throwable error) {
                        System.err.println("Error: " + error.getMessage());
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("Sequence complete.");
                    }
                });
        System.out.println("DEMO 2 ---------------");
        Observable.just(1, 2, 3)
                .subscribe(new Observer<Integer>() {

                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        System.out.println("onSubscribe ! + " + d.isDisposed());
                    }

                    @Override
                    public void onNext(Integer item) {
                        System.out.println("Next: " + item);
                    }

                    @Override
                    public void onError(Throwable error) {
                        System.err.println("Error: " + error.getMessage());
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("Sequence complete.");
                    }
                });
        System.out.println("DEMO 3 ---------------");
        Integer[] items = {0, 1, 2, 3, 4, 5};
        Observable myObservable = Observable.fromArray(items);
        myObservable.subscribe(next -> System.out.println(next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
        System.out.println("DEMO 4 ---------------");
        Integer[] items2 = {0, 1, null, 3, 4, 5};
        myObservable = Observable.fromArray(items2);
        myObservable.subscribe(next -> System.out.println(next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
        System.out.println("DEMO 5 ---------------");
        Integer[] items3 = {0, 1, 2, 3, 4, 5};
        myObservable = Observable.fromIterable(Arrays.asList(items3));
        myObservable.subscribe(next -> System.out.println(next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
        System.out.println("DEMO 6 ---------------");
        FutureTask<Long> task = new FutureTask<>(() -> System.currentTimeMillis());
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        executorService.submit(task);
        try {
            System.out.println("get:" + task.get());
        } catch (Exception e) {
            e.printStackTrace();
        }
        myObservable = Observable.fromFuture(task);
        myObservable.subscribe(next -> System.out.println("RECIEVE: " + next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
        System.out.println("DEMO 7 ---------------");
        FutureTask<Long> task2 = new FutureTask<>(() -> {
            TimeUnit.SECONDS.sleep(3);
            return System.currentTimeMillis();
        });
        myObservable = Observable.fromFuture(task2, 1, TimeUnit.SECONDS);
        executorService.submit(task2);
        myObservable.subscribe(next -> System.out.println("RECIEVE WITHIN 1s: " + next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
        System.out.println("DEMO 8 ---------------");
        task2 = new FutureTask<>(() -> {
            TimeUnit.SECONDS.sleep(3);
            return System.currentTimeMillis();
        });
        myObservable = Observable.fromFuture(task2, 4, TimeUnit.SECONDS);
        executorService.submit(task2);
        myObservable.subscribe(next -> System.out.println("RECIEVE WITHIN 4s: " + next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
        System.out.println("DEMO 9 ---------------");
        task2 = new FutureTask<>(() -> {
            TimeUnit.SECONDS.sleep(3);
            return System.currentTimeMillis();
        });
        myObservable = Observable.fromFuture(task2, 1, TimeUnit.SECONDS);
        executorService.submit(task2);
        myObservable.subscribe(next -> System.out.println("RECIEVE WITHIN 1s: " + next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
        System.out.println("DEMO 10 ---------------");
        task2 = new FutureTask<>(() -> {
            TimeUnit.SECONDS.sleep(3);
            return System.currentTimeMillis();
        });
        myObservable = Observable.fromFuture(task2, 4, TimeUnit.SECONDS);
        executorService.submit(task2);
        myObservable.subscribe(next -> System.out.println("RECIEVE WITHIN 4s: " + next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
        executorService.shutdown();
    }
}
